package com.wzy.gmail.realtime.app

import com.wzy.gmail.realtime.util.{MyBeanUtils, MyEsUtils, MyKafkaUtils, MyOffsetUtils, MyRedisUtils}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import com.alibaba.fastjson.{JSON, JSONObject}
import com.wzy.gmail.realtime.bean.{DauInfo, PageLog}
import redis.clients.jedis.{Jedis, Pipeline}

import java.text.SimpleDateFormat
import java.time.{LocalDate, Period}
import java.{lang, util}
import java.util.Date
import scala.collection.mutable.ListBuffer


/**
 * 日活宽表
 * 1.准备实时环境
 * 2.从redis中读取偏移量
 * 3.从kafka中消费数据
 * 4.提取偏移量结束点
 * 5.处理数据
 * 5.1 转换数据结构
 * 5.2 去重
 * 5.3 维度关联
 * 6.写入ES
 * 7.提交offset
 */
object DwdDauApp {
  def main(args: Array[String]): Unit = {
    // 0.状态还原
    revertState()
    //1.准备实时环境
    val sparkConf: SparkConf = new SparkConf().setAppName("dwd_dau_app").setMaster("local[4]")
    val ssc = new StreamingContext(sparkConf, Seconds(5))

    //2.从redis中提取偏移量
    val topicName: String = "DWD_PAGE_LOG_TOPIC_1018"
    val groupId: String = "DWD_DAU_GROUP_1018"
    val offsets: Map[TopicPartition, Long] = MyOffsetUtils.readOffset(topicName, groupId)

    var kafkaDStream: InputDStream[ConsumerRecord[String, String]] = null
    //3.从kafka中消费数据
    if (offsets != null && offsets.nonEmpty) {
      kafkaDStream = MyKafkaUtils.getKafkaDStream(topicName, ssc, groupId, offsets)
    }
    else {
      kafkaDStream = MyKafkaUtils.getKafkaDStream(topicName, ssc, groupId)
    }
    //4.提取offset结束点
    var offsetRanges: Array[OffsetRange] = null
    val offsetRangesDStream: DStream[ConsumerRecord[String, String]] = kafkaDStream.transform(
      rdd => {
        offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
        rdd
      }
    )
    //5.处理数据
    //5.1转换结构
    val pageLogDStream: DStream[PageLog] = offsetRangesDStream.map(
      consumerRecord => {
        val value: String = consumerRecord.value()
        val pageLog: PageLog = JSON.parseObject(value, classOf[PageLog])
        pageLog
      }
    )
    pageLogDStream.cache()
    pageLogDStream.foreachRDD(
      rdd => println("自我审查前:" + rdd.count())
    )
    //5.2去重
    // 自我审查：将页面访问数据中last_page_id不为空的数据过滤掉
    val filterDStream: DStream[PageLog] = pageLogDStream.filter(
      pageLog => pageLog.last_page_id == null
    )
    filterDStream.foreachRDD(
      rdd => {
        println("自我审查后:" + rdd.count())
        println("----------------------------")
      }
    )
    // 第三方审查：通过redis将当日活跃的mid维护起来，自我审查后的每条数据需要到redis中进行比对去重
    //redis中如何维护日活状态
    //类型 list/set
    //key:  DAU:DATE
    //value:  mid部分
    //写入api： lpush/rpush   sadd
    //读取api:  lrange        smembers
    //过期: 24h
    //    filterDStream.filter() //每条数据执行一次，redis连接频繁
    val redisFilterDStream: DStream[PageLog] = filterDStream.mapPartitions(
      pageLogIter => {
        val pageLogList: List[PageLog] = pageLogIter.toList
        println("第三方审查前:" + pageLogList.size)

        // 存储要的数据
        val pageLogs: ListBuffer[PageLog] = ListBuffer[PageLog]()
        val jedis: Jedis = MyRedisUtils.getJedisFromPool()
        val sdf = new SimpleDateFormat("yyyy-MM-dd")

        for (pageLog <- pageLogList) {
          //提取每条数据中的mid(日活的统计基于mid，也可以基于uid)
          val mid: String = pageLog.mid

          //获取日期，测试不同天数据，不能直接获取系统时间
          val ts: Long = pageLog.ts
          val date = new Date(ts)
          val dateStr: String = sdf.format(date)
          val redisDauKey: String = s"DAU:$dateStr"

          //redis判断是否有mid
          //下面代码在分布式环境中，存在并发问题，可能多个并行度同时进入到if中，导致最终保留多条同一mid的数据
          //list
          //          val mids: util.List[String] = jedis.lrange(redisDauKey, 0, -1)
          //          if(!mids.contains(mid)){
          //            jedis.lpush(redisDauKey,mid)
          //            pageLogs.append(pageLog)
          //          }
          val isNew: lang.Long = jedis.sadd(redisDauKey, mid) // 判断写入的原子操作
          if (isNew == 1L) {
            pageLogs.append(pageLog)
          }
        }
        jedis.close()
        println("第三方审查后:" + pageLogList.size)
        pageLogs.iterator
      }
    )
    //    redisFilterDStream.print()

    // 5.3 维度关联
    val dauInfoDStreams: DStream[DauInfo] = redisFilterDStream.mapPartitions(
      pageLogIter => {
        val dauInfos: ListBuffer[DauInfo] = ListBuffer[DauInfo]()
        val sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
        val jedis: Jedis = MyRedisUtils.getJedisFromPool()
        for (pageLog <- pageLogIter) {
          val dauInfo = new DauInfo()
          // 1.将pageLog中以后字段拷贝到DauInfo中
          // 将pageLog中的每个字段挨个提取，复制给dauInfo中对应的字段(不方便)
          // 通过对象拷贝来完成
          MyBeanUtils.copyProperties(pageLog, dauInfo)
          // 2.补充维度
          // 2.1 用户信息维度
          val uid: String = pageLog.user_id
          val redisUidKey: String = s"DIM:USER_INFO:$uid"
          val userInfoJson: String = jedis.get(redisUidKey)
          val userInfoJsonObj: JSONObject = JSON.parseObject(userInfoJson)
          // 提取性别
          val gender: String = userInfoJsonObj.getString("gender")
          // 提取生日
          val birthday: String = userInfoJsonObj.getString("birthday") // 1972-03-22
          // 换算年龄
          val birthdayLd: LocalDate = LocalDate.parse(birthday)
          val nowLd: LocalDate = LocalDate.now()
          val period: Period = Period.between(birthdayLd, nowLd)
          val age: Int = period.getYears
          // 补充到对象中
          dauInfo.user_gender = gender
          dauInfo.user_age = age.toString

          // 2.2 地区信息维度
          // redis中，DIM:BASE:PROVINCE:1 DIMlBASE_PROVINCE:11000
          val province_id: String = dauInfo.province_id
          val redisProvinceKey: String = s"DIM:BASE_PROVINCE:$province_id"
          val provinceJson: String = jedis.get(redisProvinceKey)
          val provinceJsonObj: JSONObject = JSON.parseObject(provinceJson)

          val provinceName: String = provinceJsonObj.getString("name")
          val provinceIsoCode: String = provinceJsonObj.getString("iso_code")
          val province3166: String = provinceJsonObj.getString("iso_3166_2")
          val provinceAreaCode: String = provinceJsonObj.getString("area_code")

          //补充到对象中
          dauInfo.province_name = provinceName
          dauInfo.province_iso_code = provinceIsoCode
          dauInfo.province_3166_2 = province3166
          dauInfo.province_area_code = provinceAreaCode

          // 2.3 日期字段处理
          val date = new Date(pageLog.ts)
          val dtHr: String = sdf.format(date)
          val dtHrArr: Array[String] = dtHr.split(" ")
          val dt: String = dtHrArr(0)
          val hr: String = dtHrArr(1).split(":")(0)
          // 补充到对象中
          dauInfo.dt = dt
          dauInfo.hr = hr
          dauInfos.append(dauInfo)
        }
        jedis.close()
        dauInfos.iterator
      }
    )
    //    dauInfoDStreams.print(100)
    // 写入到ES中
    // 按照天分割索引，通过索引模板控制mapping，setting，aliases
    // 批量读写,准备ES工具类
    dauInfoDStreams.foreachRDD(
      rdd => {
        rdd.foreachPartition(
          dauInfoIter => {
            val docs: List[(String, DauInfo)] = dauInfoIter.map(dauInfo => (dauInfo.mid, dauInfo)).toList
            if (docs.nonEmpty) {
              //索引名
              //真实环境直接用当前日期
              //模拟数据，会生成不同天的数据
              //从第一条数据获取ts
              val head: (String, DauInfo) = docs.head
              val ts: Long = head._2.ts
              val sdf: SimpleDateFormat = new SimpleDateFormat("yyyy-MM-dd")
              val dataStr: String = sdf.format(new Date(ts))
              val indexName: String = s"gmall_dau_info_1018_$dataStr"
              // 写入es
              MyEsUtils.bulkSave(indexName, docs)
            }
          }
        )
        // 提交offset
        MyOffsetUtils.saveOffset(topicName, groupId, offsetRanges)
      }
    )
    ssc.start()
    ssc.awaitTermination()
  }

  /**
   * 状态还原
   *
   * 在每次启动任务时，进行一次状态还原，以es为准，将所有mid提取出来，覆盖到redis中
   */
  def revertState(): Unit = {
    // 从ES中查询到所有的mid
    val date: LocalDate = LocalDate.now()
    val indexName: String = s"gmall_dau_info_1018_$date"
    val fieldName: String = "mid"
    val mids: List[String] = MyEsUtils.searchField(indexName, fieldName)

    // 删除redis中记录的状态（所有的mid）
    val jedis: Jedis = MyRedisUtils.getJedisFromPool()
    val jedisDauKey: String = s"DAU:$date"
    jedis.del(jedisDauKey)

    // 将从es中查询到的mid覆盖到redis中
    if (mids != null && mids.nonEmpty) {
      //      for (mid <- mids) {
      //        jedis.sadd(jedisDauKey,mid)
      //      }
      val pipeline: Pipeline = jedis.pipelined()
      for (mid <- mids) {
        pipeline.sadd(jedisDauKey, mid)
      }
      pipeline.sync()
    }
    jedis.close()
  }

}
